# -*- coding: utf-8 -*-
"""
Kafka WebSocket消费者模块

该模块实现了基于Django Channels的WebSocket消费者，用于实时接收Kafka消息并推送到前端。
主要功能：
1. 建立WebSocket连接
2. 启动独立的Kafka消费者线程
3. 实时接收Kafka消息并通过WebSocket推送到前端
4. 处理连接断开和异常情况
"""

import json
import asyncio
import threading
import logging
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async
from kafka import KafkaConsumer as KafkaPyConsumer
from .models import KafkaConfig

# 配置日志记录器
logger = logging.getLogger(__name__)

class KafkaConsumer(AsyncWebsocketConsumer):
    """
    Kafka WebSocket消费者类
    
    继承自AsyncWebsocketConsumer，处理WebSocket连接和Kafka消息消费。
    每个WebSocket连接对应一个Kafka配置，启动独立的消费者线程。
    """
    
    def __init__(self, *args, **kwargs):
        """初始化消费者实例"""
        super().__init__(*args, **kwargs)
        self.kafka_thread = None  # Kafka消费者线程
        self.kafka_consumer = None  # Kafka消费者实例
        self.stop_event = threading.Event()  # 线程停止事件

    async def connect(self):
        """
        处理WebSocket连接建立
        
        流程：
        1. 从URL路径中提取Kafka配置ID
        2. 验证配置是否存在
        3. 接受WebSocket连接
        4. 启动独立的Kafka消费者线程
        5. 记录连接建立日志
        """
        # 从WebSocket URL路径中获取配置ID参数
        self.config_id = self.scope['url_route']['kwargs']['config_id']
        
        try:
            # 异步检查Kafka配置是否存在（防止无效配置ID）
            config = await self.get_kafka_config(self.config_id)
            
            # 接受WebSocket连接请求
            await self.accept()
            
            # 创建并启动Kafka消费者线程
            # 使用独立线程避免阻塞WebSocket的异步事件循环
            self.kafka_thread = threading.Thread(
                target=self.kafka_consumer_thread, 
                args=(self.config_id,)
            )
            self.kafka_thread.daemon = True  # 设置为守护线程，主程序退出时自动结束
            self.kafka_thread.start()
            
            logger.info(f"WebSocket连接已建立，配置ID: {self.config_id}")
        except KafkaConfig.DoesNotExist:
            # 处理Kafka配置不存在的情况
            await self.accept()  # 先接受连接才能发送错误信息
            error_msg = {'error': f'未找到ID为{self.config_id}的Kafka配置'}
            await self.send(text_data=json.dumps(error_msg))
            await self.close()  # 发送错误信息后关闭连接
            logger.error(f"未找到ID为{self.config_id}的Kafka配置")
        except Exception as e:
            # 处理其他异常情况
            # 先尝试接受连接再发送错误信息（如果连接还没建立）
            try:
                await self.accept()
            except:
                pass  # 如果连接已建立或出错，忽略异常
                
            error_msg = {'error': f'WebSocket连接出错: {str(e)}'}
            try:
                await self.send(text_data=json.dumps(error_msg))
            except:
                pass  # 如果发送失败，忽略异常
                
            await self.close()
            logger.error(f"WebSocket连接出错: {str(e)}")

    async def disconnect(self, close_code):
        """
        处理WebSocket连接断开
        
        清理流程：
        1. 设置线程停止事件
        2. 关闭Kafka消费者连接
        3. 等待消费者线程结束
        4. 调用父类的断开连接方法
        
        Args:
            close_code: WebSocket关闭代码
        """
        logger.info(f"WebSocket连接断开，配置ID: {self.config_id}")
        
        # 设置停止事件，通知Kafka消费者线程退出
        self.stop_event.set()
        
        # 安全关闭Kafka消费者连接
        if self.kafka_consumer:
            try:
                self.kafka_consumer.close()
            except Exception as e:
                logger.error(f"关闭Kafka消费者时出错: {str(e)}")
        
        # 等待消费者线程正常结束
        if self.kafka_thread and self.kafka_thread.is_alive():
            self.kafka_thread.join(timeout=5)  # 设置5秒超时避免无限等待
            
        # 调用父类的断开连接方法
        await super().disconnect(close_code)

    @database_sync_to_async
    def get_kafka_config(self, config_id):
        """
        异步获取Kafka配置
        
        使用database_sync_to_async装饰器将同步的Django ORM操作转换为异步操作，
        避免在异步上下文中直接调用同步数据库操作导致的错误。
        
        Args:
            config_id: Kafka配置的主键ID
            
        Returns:
            KafkaConfig: Kafka配置对象
            
        Raises:
            KafkaConfig.DoesNotExist: 当指定ID的配置不存在时
        """
        return KafkaConfig.objects.get(id=config_id)
    
    def kafka_consumer_thread(self, config_id):
        """
        Kafka消费者线程函数
        
        在独立线程中运行Kafka消费者，持续监听指定主题的消息。
        接收到消息后通过WebSocket发送到前端客户端。
        
        Args:
            config_id: Kafka配置的主键ID
            
        线程安全说明：
        - 在独立线程中运行，不会阻塞WebSocket的异步事件循环
        - 使用threading.Event()进行线程间通信
        - 使用async_to_sync()安全地调用异步WebSocket发送方法
        """
        try:
            # 在线程中使用同步方式获取Kafka配置
            config = KafkaConfig.objects.get(id=config_id)
            
            # 创建Kafka消费者实例
            # 配置说明：
            # - topic: 监听的主题名称
            # - bootstrap_servers: Kafka服务器地址列表
            # - value_deserializer: 消息值反序列化器，将字节数据转换为JSON对象
            # - auto_offset_reset: 当没有初始偏移量时的重置策略，'latest'表示从最新消息开始
            # - enable_auto_commit: 启用自动提交偏移量
            # - group_id: 消费者组ID，每个配置使用独立的组避免冲突
            # - reconnect_backoff_max_ms: 重连最大等待时间
            # - request_timeout_ms: 请求超时时间
            self.kafka_consumer = KafkaPyConsumer(
                config.topic,
                bootstrap_servers=[config.url],
                value_deserializer=lambda m: json.loads(m.decode('utf-8')) if m else None,
                auto_offset_reset='latest',  # 从最新消息开始消费
                enable_auto_commit=True,
                group_id=f'kafka_tool_group_{config_id}',  # 每个配置使用独立的消费者组
                # consumer_timeout_ms=1000,  # 已注释：移除消费者超时，避免频繁重新平衡
                reconnect_backoff_max_ms=10000,  # 最大重连等待时间
                request_timeout_ms=30000,  # 请求超时时间
                # 禁用压缩相关配置，避免snappy压缩库依赖问题
                fetch_max_wait_ms=500,  # 减少等待时间
                max_partition_fetch_bytes=1048576,  # 1MB
                # 如果服务器使用压缩，客户端会自动处理，但不依赖特定压缩库
                check_crcs=False,  # 禁用CRC检查以提高兼容性
                # 针对Kafka 2.5.1的snappy压缩问题，添加更多兼容性配置
                api_version=(2, 5, 1),  # 明确指定Kafka版本
                security_protocol='PLAINTEXT',  # 明确指定安全协议
                # 设置较小的fetch size以避免压缩问题
                fetch_min_bytes=1,
                fetch_max_bytes=52428800,  # 50MB
                # 禁用某些可能导致压缩问题的特性
                exclude_internal_topics=True
            )
            
            logger.info(f"Kafka消费者已启动，监听配置ID: {config_id}, 主题: {config.topic}")
            
            # 持续消费消息循环
            # 使用for循环遍历消费者，这是一个阻塞操作，会持续等待新消息
            for message in self.kafka_consumer:
                # 检查线程停止事件，如果设置则退出循环
                if self.stop_event.is_set():
                    logger.info("收到停止信号，退出消费者循环")
                    break
                
                # 构造标准化的消息数据结构
                # 包含消息的元数据和实际内容
                message_data = {
                    'topic': message.topic,        # 主题名称
                    'partition': message.partition, # 分区号
                    'offset': message.offset,       # 消息偏移量
                    'timestamp': message.timestamp, # 消息时间戳
                    'value': message.value          # 消息内容（已反序列化为JSON）
                }
                
                logger.info(f"收到Kafka消息: {message_data}")
                
                # 将消息发送到WebSocket客户端
                try:
                    # 导入必要的异步同步转换模块
                    import asyncio
                    from asgiref.sync import async_to_sync
                    
                    # 使用async_to_sync将异步的self.send方法转换为同步调用
                    # 这样可以在同步的线程中安全地调用异步的WebSocket发送方法
                    # text_data参数需要是JSON字符串格式
                    async_to_sync(self.send)(text_data=json.dumps(message_data))
                    logger.info(f"消息已发送到WebSocket客户端")
                except Exception as e:
                    logger.error(f"发送消息到WebSocket时出错: {str(e)}")
                    break  # 发送失败时退出循环，避免继续处理消息
                    
        except KafkaConfig.DoesNotExist:
            # 处理Kafka配置不存在的异常
            logger.error(f"未找到ID为{config_id}的Kafka配置")
            # 尝试发送错误信息到WebSocket客户端
            try:
                from asgiref.sync import async_to_sync
                error_data = {'error': f'未找到ID为{config_id}的Kafka配置'}
                async_to_sync(self.send)(text_data=json.dumps(error_data))
            except Exception as e:
                logger.error(f"发送错误信息到WebSocket时出错: {str(e)}")
        except Exception as e:
            # 处理其他所有异常（如Kafka连接失败、网络错误等）
            logger.error(f"Kafka消费者线程出错: {str(e)}")
            # 尝试发送错误信息到WebSocket客户端
            try:
                from asgiref.sync import async_to_sync
                error_data = {'error': f'Kafka消费者出错: {str(e)}'}
                async_to_sync(self.send)(text_data=json.dumps(error_data))
            except Exception as send_error:
                logger.error(f"发送错误信息到WebSocket时出错: {str(send_error)}")